package com.bigdata;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class HBaseScanDemo {

    private Configuration conf;
    // HBase 的连接对象
    private Connection conn;
    // HBase 操作DDL的对象
    private Admin admin;

    @Before
    public void init() throws Exception {
        conf = HBaseConfiguration.create();
        // 设置zk的连接信息
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    /**
     * 分页过滤器
     */
    @Test
    public void pageFilterTest() throws Exception {
        // scan 'ns1:t1', {COLUMNS => ['c1', 'c2'], LIMIT => 10, STARTROW => 'xyz'}
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        // 想要用代码分页查询所有的数据
        Scan scan = new Scan();

        // scan.withStartRow("1500100010".getBytes());

        // 分页过滤器，默认从第一页开始查
        Long pageSize = 10L;
        PageFilter filter = new PageFilter(pageSize);
        // 追加Filter 过滤器
        scan.setFilter(filter);

        // 默认扫描不到数据，跳出循环
        boolean flag;

        // 起始、用来追加的数组
        byte[] initBys = new byte[]{0x00};
        // 记录上一页的最后一个rowkey的字节数组
        byte[] rowkeyBys = null;

        while (true) {

            // 非第一页的时候，rowkeyBys就不是null了
            if (rowkeyBys != null) {
                // withStartRow是包含这一样数据，需要在默认追加一些“00”，可以做到不取这一行数据
                scan.withStartRow(Bytes.add(rowkeyBys, initBys));
            }

            // 发送scan命令，扫描一页数据
            ResultScanner scanner = table.getScanner(scan);

            // 重置为true，因为上一页有数据，flag已经变味了false
            flag = true;
            Result result;
            int count = 0;
            while ((result = scanner.next()) != null) {
                processResult(result);
                count++;
                // 改为false，说明扫描到了数据
                flag = false;

                // 取最后一个条数据rowkey并赋值
                rowkeyBys = result.getRow();
            }
            System.out.println("此次扫描到了 " + count + " 行数据");

            if (flag) {
                break;
            }


        }


        // 关闭资源
        table.close();
    }


    /**
     * 专用过滤器
     */
    @Test
    public void scanTest2() throws Exception {
        // scan 'api_test:student'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Scan scan = new Scan();

        // 单列列值过滤器
//        SingleColumnValueFilter filter = new SingleColumnValueFilter("info".getBytes(), "name".getBytes(),
//                CompareOperator.EQUAL,
//                new BinaryComparator("吕金鹏".getBytes()));

        // 排除name列
//        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter("info".getBytes(), "name".getBytes(),
//                CompareOperator.EQUAL,
//                new BinaryComparator("吕金鹏".getBytes()));

//        PrefixFilter filter = new PrefixFilter(Bytes.toBytes("1500100010"));

        // 列名前缀过滤器
        // ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("na"));

        // 时间戳过滤器
//        List<Long> list = new ArrayList<>();
//        // 多个值之间组成时间范围
//        list.add(1705545530830L);
//        list.add(1705545530835L);
//
//        TimestampsFilter filter = new TimestampsFilter(list);

        FirstKeyOnlyFilter filter = new FirstKeyOnlyFilter();

        // 追加Filter 过滤器
        scan.setFilter(filter);

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);

        processScanner(scanner);

        // 关闭资源
        table.close();
    }


    /**
     * scan 扫描一批数据
     */
    @Test
    public void scanTest1() throws Exception {
        // scan 'api_test:student'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Scan scan = new Scan();

        // 追加扫描条件
//        scan.withStartRow("1500100011".getBytes());
//        scan.withStopRow("1500100018".getBytes());

        // 创建过滤器
        /*
        确定3点：
        1.使用的是哪个Filter，表的比较逻辑
        2.需要确定比较符
        3.确定值的比较逻辑
         */
        // FamilyFilter filter = new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator("info".getBytes()));
        // FamilyFilter filter = new FamilyFilter(CompareOperator.EQUAL, new BinaryPrefixComparator("inf".getBytes()));

        // name,age的值可以查到吗， select c1,c2
        // QualifierFilter filter = new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator("clazz".getBytes()));

        // ValueFilter ：基于单元格 (cell) 的值来过滤数据，会取所有的列的值来匹配，只会返回匹配到的列数据；
        // ValueFilter filter = new ValueFilter(CompareOperator.EQUAL, new BinaryPrefixComparator("文科".getBytes()));


        // DependentColumnFilter ：指定一个参考列来过滤其他列的过滤器，过滤的原则是基于参考列的时间戳来进行筛选 。
        // 会把所有的列都返回
        DependentColumnFilter filter = new DependentColumnFilter("info".getBytes(), "clazz".getBytes(), false,
                CompareOperator.EQUAL, new BinaryPrefixComparator("文科".getBytes()));


        // 追加Filter 过滤器
        scan.setFilter(filter);

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);

        processScanner(scanner);

        // 关闭资源
        table.close();
    }

    /**
     * scan 扫描一批数据
     */
    @Test
    public void scanTest() throws Exception {
        // scan 'api_test:student'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Scan scan = new Scan();

        // 追加扫描条件
//        scan.withStartRow("1500100011".getBytes());
//        scan.withStopRow("1500100018".getBytes());

        // 创建过滤器
        /*
        确定3点：
        1.使用的是哪个Filter，表的比较逻辑
        2.需要确定比较符
        3.确定值的比较逻辑
         */
        // RowFilter filter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator("150010001"));

//        RowFilter filter = new RowFilter(CompareOperator.GREATER_OR_EQUAL, new BinaryComparator("1500100013".getBytes()));

        // row可以的前缀过滤
        // RowFilter filter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator("150010001".getBytes()));
        RowFilter filter = new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("^15.*0$"));


        // 追加Filter 过滤器
        scan.setFilter(filter);

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);

        processScanner(scanner);

        // 关闭资源
        table.close();
    }


    /**
     * 打印扫描结果
     *
     * @param scanner
     */
    private static void processScanner(ResultScanner scanner) {
        try {
            Result result;
            int count = 0;
            while ((result = scanner.next()) != null) {
                processResult(result);
                count++;
            }
            System.out.println("此次扫描到了 " + count + " 行数据");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 打印一个result对象中的数据
     *
     * @param result
     */
    private static void processResult(Result result) {
        // 获取rowkey
        String rowkey = Bytes.toString(result.getRow());

//        String name = new String(result.getValue("info".getBytes(), "name".getBytes()));
        // 列不存在，不会出现空指针
        String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

        System.out.println("id:" + rowkey + "---name:" + name + "---age:" + age
                + "---gender:" + gender + "---clazz:" + clazz);
    }


    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }
}
